package cn.wubo.multi.thread.demo03;

import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.transaction.PlatformTransactionManager;
import org.springframework.transaction.TransactionDefinition;
import org.springframework.transaction.TransactionStatus;
import org.springframework.transaction.support.DefaultTransactionDefinition;

import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * Spring事务管理器PlatformTransactionManager
 * 多线程事务回滚
 */
@Slf4j
@Service
public class Demo0303 {

    @Autowired
    public PlatformTransactionManager transactionManager;

    public void asyncSaveBatch() {
        // 线程个数
        final int SIZE = 6;
        // 交给单个线程处理，失败则加1
        final AtomicInteger failedCounter = new AtomicInteger();

        final CountDownLatch latch = new CountDownLatch(SIZE);

        final ExecutorService pool = Executors.newFixedThreadPool(SIZE);

        List<TransactionStatus> transactionStatuses = Collections.synchronizedList(new ArrayList<TransactionStatus>());

        long beginTime = System.currentTimeMillis();

        for (int i = 0; i < SIZE; i++) {
            pool.execute(() -> {
                final boolean success = Math.random() > 0.1;
                log.info(Thread.currentThread().getName() + ", success = " + success);
                // 将每个线程中的事务状态存入集合单当中
                DefaultTransactionDefinition def = new DefaultTransactionDefinition();
                // 事物隔离级别，开启新事务，这样会比较安全些。
                def.setPropagationBehavior(TransactionDefinition.PROPAGATION_REQUIRES_NEW);
                // 获得事务状态
                TransactionStatus status = transactionManager.getTransaction(def);
                transactionStatuses.add(status);

                try {
                    /* 业务逻辑 */
                } catch (Exception e) {
                    failedCounter.getAndIncrement();
                } finally {
                    transactionManager.commit(status);
                }
                latch.countDown();
            });
        }

        try {
            latch.await();
            if (failedCounter.get() > 0) {
                for (TransactionStatus transactionStatus : transactionStatuses) {
                    transactionStatus.setRollbackOnly();
                }
                log.info("执行了回滚过程，回滚事物数据集合大小" + transactionStatuses.size());
            }
            long endTime = System.currentTimeMillis();
            log.info("执行批量插入数据耗时" + (int) ((endTime - beginTime)) / 1000 + "秒");
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

    }
}
